package olap4cloud.hbase;

import java.io.File;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.util.ArrayList;
import java.util.concurrent.CompletionService;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.HTablePool;
import org.apache.log4j.Logger;
import org.apache.log4j.xml.DOMConfigurator;

public class MainDriver {

	/**
	 * @uml.property  name="pool"
	 * @uml.associationEnd  multiplicity="(1 1)"
	 */
	private HTablePool pool;
	/**
	 * @uml.property  name="conf"
	 * @uml.associationEnd  multiplicity="(1 1)"
	 */
	private Configuration conf;

	private static int MAXPOOLHBASE = 20;
	private static int MAXTHREADPOOL = 500;

	private static Logger log1 = Logger.getLogger("logger");

	public MainDriver() {
		conf = HBaseConfiguration.create(new Configuration());
		pool = new HTablePool(conf, MAXPOOLHBASE);
	}

	/**
	 * Create hbase pool
	 * 
	 * @param pool
	 */
	public void initHBasePool(HTablePool pool) {
		
		HTableInterface[] tables = new HTableInterface[MAXPOOLHBASE];
		for (int n = 0; n < MAXPOOLHBASE; n++) {
			tables[n] = pool.getTable(TrackingTable.getTableName());
		}
		for (int n = 0; n < MAXPOOLHBASE; n++) {
			try {
				pool.putTable(tables[n]);
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
	}

	/**
	 * Put result of job hadoop to hbase
	 * 
	 * @throws FileNotFoundException
	 */
	public boolean putDataToHbase(String fileName) throws FileNotFoundException {

		initHBasePool(pool);
		ArrayList<File> listFile = new ArrayList<File>();
		File f = new File(fileName);
		if (!f.isDirectory())
			listFile.add(f);
		else {
			for (File tmp : f.listFiles()) {
				listFile.add(tmp);
			}
		}
		System.out.println("Number of sub file : " + listFile.size());
		long startTime = System.currentTimeMillis();
		for (File tmp : listFile) {
			ExecutorService eservice = Executors
					.newFixedThreadPool(MAXTHREADPOOL);
			CompletionService<Object> objectCompletionService = new ExecutorCompletionService<Object>(
					eservice);
			DataToHbase task = new DataToHbase(pool, tmp.getPath());
			task.setEservice(eservice);
			task.setObjectCompletionService(objectCompletionService);
			try {
				task.call();
			} catch (Exception e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
				return false;
			}
			eservice.shutdown();
			while (!eservice.isTerminated()) {
			}
		}
		long stopTime = System.currentTimeMillis();
		log1.info("Finish.Time use to put data to Hbase : "
				+ (stopTime - startTime));
		return true;
	}

	public static void main(String[] args) throws FileNotFoundException {
		DOMConfigurator.configure("log4j.xml");
		MainDriver statistic = new MainDriver();
		if (args.length > 0) {
			statistic.putDataToHbase(args[0]);
			System.out.println("Finish put data---------------------");
			statistic.putDataToHbase(args[0]);
			System.out.println("Finish statistic data---------------------");
		} else {
			statistic.putDataToHbase("News-1366128000000");
			System.out.println("Finish put data---------------------");
		}
	}
}
